package com.xinjian.jianba.es.pool;

import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import java.util.concurrent.atomic.AtomicReference;


public class ElasticSearchClientFactory implements PooledObjectFactory<RestHighLevelClient> {

    private AtomicReference<HostAndPort> nodesReference = new AtomicReference<HostAndPort>();
    public ElasticSearchClientFactory(HostAndPort clusterNodes){
        this.nodesReference.set(clusterNodes);
    }
    @Override
    public PooledObject<RestHighLevelClient> makeObject() throws Exception {
        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
    	credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(nodesReference.get().getUserName(), nodesReference.get().getPassword()));
    	RestClientBuilder clientBuilder = RestClient.builder(new HttpHost(nodesReference.get().getHost(), nodesReference.get().getPort(), "http")).setHttpClientConfigCallback(httpClientBuilder ->
        httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
        
        RestHighLevelClient client = new RestHighLevelClient(clientBuilder);
        return new DefaultPooledObject<RestHighLevelClient>(client);
    }
    @Override
    public void destroyObject(PooledObject<RestHighLevelClient> pooledObject) throws Exception {
        RestHighLevelClient client = pooledObject.getObject();
        if(client!=null&&client.ping()){
            try {
                client.close();
            }catch (Exception e){
                //ignore
            }
        }
    }
    @Override
    public boolean validateObject(PooledObject<RestHighLevelClient> pooledObject) {
        RestHighLevelClient client = pooledObject.getObject();
        try {
            return client.ping();
        }catch(Exception e){
            return false;
        }
    }

    @Override
    public void activateObject(PooledObject<RestHighLevelClient> pooledObject) throws Exception {
        RestHighLevelClient client = pooledObject.getObject();
        boolean response = client.ping();
    }

    @Override
    public void passivateObject(PooledObject<RestHighLevelClient> pooledObject) throws Exception {
        //nothing
    }


}
